/*
 * Copyright © 2015-2026 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.dongxing.springbootrocketmqstarter.core;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageQueue;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Component;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.util.Map;
import java.util.Objects;
import java.util.Optional;

/**
 * <pre>
 * 名称: RockMQTemplate
 * 描述: RockMQTemplate
 * </pre>
 * @author dongxing
 */
@Slf4j
public class RocketMQTemplate implements InitializingBean, DisposableBean {

    @Getter
    @Setter
    private DefaultMQProducer producer;

    @Getter
    @Setter
    private RocketMQSerializer serializer = new FastJsonRocketMQSerializer();

    @Getter
    @Setter
    private String charset = "UTF-8";

    /**
     * 同步发送消息
     * @param message 消息体
     * @return sendResult
     */
    public SendResult syncSend(final Message message) {
        return this.syncSend(message, producer.getSendMsgTimeout());
    }

    /**
     * 同步发送消息
     * @param message 消息体
     * @param timeout 超时时间
     * @return sendResult
     */
    public SendResult syncSend(final Message message, final long timeout){
        try {
            SendResult sendResult = producer.send(message, timeout);
            return sendResult;
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 同步发送消息
     *
     * @param topic  the topic
     * @param tag    the tag
     * @param msgObj 消息对象
     * @param properties the properties
     * @return sendResult
     */
    public SendResult syncSend(final String topic, final String tag, final Object msgObj,
                               final Map<String, Object> properties) {
        return this.syncSend(topic, tag, msgObj, properties, this.producer.getSendMsgTimeout());
    }

    /**
     * 同步发送消息
     *
     * @param topic the topic
     * @param tag   the tag
     * @param msgObj 消息体
     * @param properties the properties
     * @param timeout 超时时间
     * @return sendResult
     */
    public SendResult syncSend(final String topic, final String tag, final Object msgObj,
                               final Map<String, Object> properties, final long timeout) {
        Message message = createMessage(topic, tag, msgObj, properties);
        return this.syncSend(message, timeout);
    }

    /**
     * 异步发送消息
     * @param message 消息体
     * @param sendCallback 回调方法
     */
    public void asyncSend(final Message message, final SendCallback sendCallback) {
        this.asyncSend(message, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     *异步发送消息
     * @param message 消息体
     * @param sendCallback 回调方法
     * @param timeout 超时
     */
    public void asyncSend(final Message message, final SendCallback sendCallback, final long timeout) {
        try {
            producer.send(message, sendCallback, timeout);
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 异步发送消息
     *
     * @param topic the topic
     * @param tag   the tag
     * @param msgObj 消息体
     * @param properties the propeties
     * @param sendCallback 回调方法
     */
    public void asyncSend(final String topic, final String tag, final Object msgObj,
            final Map<String, Object> properties, final SendCallback sendCallback) {
        this.asyncSend(topic, tag, msgObj, properties, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * 异步发送消息
     *
     * @param topic the topic
     * @param tag   the tag
     * @param msgObj 消息体
     * @param properties the propeties
     * @param sendCallback 回调方法
     * @param timeout 超时时间
     */
    public void asyncSend(final String topic, final String tag, final Object msgObj,
                          final Map<String, Object> properties, final SendCallback sendCallback, final long timeout) {
        Message message = createMessage(topic, tag, msgObj, properties);
        this.asyncSend(message, sendCallback, timeout);
    }

    /**
     * 不关心消息是否送达
     * @param message 消息体
     */
    public void sendOneWay(final Message message){
        try {
            producer.sendOneway(message);
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(),  message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 不关心消息是否送达
     *
     * @param topic the topic
     * @param tag   the tag
     * @param msgObj 消息体
     * @param properties the properties
     */
    public void sendOneWay(final String topic, final String tag, final Object msgObj, final Map<String, Object> properties) {
        Message message = createMessage(topic, tag, msgObj, properties);
        sendOneWay(message);
    }

    /**
     * 同步发送到指定队列
     * @param message 消息体
     * @param messageQueue 消息队列
     * @return sendResult
     */
    public SendResult syncSend(final Message message, final MessageQueue messageQueue) {
        return this.syncSend(message, messageQueue, producer.getSendMsgTimeout());
    }

    /**
     * 同步发送到指定队列
     * @param message 消息体
     * @param messageQueue 消息队列
     * @param timeout 超时时间
     * @return sendResult
     */
    public SendResult syncSend(final Message message, final MessageQueue messageQueue, final long timeout){
        try {
            SendResult sendResult = producer.send(message, messageQueue, timeout);
            return sendResult;
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }
    /**
     * 同步发送到指定队列
     *
     * @param topic the topic
     * @param tag   the tag
     * @param msgObj 消息体
     * @param properties the propeties
     * @param mq mq
     * @return sendResult
     */
    public SendResult syncSend(final String topic, final String tag, final Object msgObj,
                               final Map<String, Object> properties, final MessageQueue mq) {
        return this.syncSend(topic, tag, msgObj, properties, mq, producer.getSendMsgTimeout());
    }

    /**
     * 同步发送到指定队列
     *
     * @param topic the topic
     * @param tag   the tag
     * @param msgObj 消息体
     * @param properties the propeties
     * @param mq    the mq
     * @param timeout 超时时间
     * @return sendResult
     */
    public SendResult syncSend(final String topic, final String tag, final Object msgObj,
                               final Map<String, Object> properties, final MessageQueue mq, final long timeout) {
        Message message = createMessage(topic, tag, msgObj, properties);
        return this.syncSend(message, mq, timeout);
    }

    /**
     * 异步发送到指定队列
     * @param message      消息体
     * @param messageQueue 消息队列
     * @param sendCallback 回调方法
     */
    public void asyncSend(final Message message, final MessageQueue messageQueue, final SendCallback sendCallback) {
        this.asyncSend(message, messageQueue, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * 异步发送到指定队列
     * @param message      消息体
     * @param messageQueue 消息队列
     * @param sendCallback 回调方法
     * @param timeout      超时时间
     */
    public void asyncSend(final Message message, final MessageQueue messageQueue, final SendCallback sendCallback,
                          final long timeout) {
        try {
            producer.send(message, messageQueue, sendCallback, timeout);
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }
    /**
     * 异步发送到指定队列
     *
     * @param topic      the topic
     * @param tag        the tag
     * @param msgObj     消息体
     * @param properties the properties
     * @param mq         the mq
     * @param sendCallback 回调方法
     */
    public void asyncSend(final String topic, final String tag, final Object msgObj,
                          final Map<String, Object> properties, final MessageQueue mq, final SendCallback sendCallback) {
        this.asyncSend(topic, tag, msgObj, properties, mq, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * 异步发送到指定队列
     * @param topic      the topic
     * @param tag        the tag
     * @param msgObj     消息体
     * @param properties the properties
     * @param mq         the mq
     * @param sendCallback 回调方法
     * @param timeout    超时时间
     */
    public void asyncSend(final String topic, final String tag, final Object msgObj,
                          final Map<String, Object> properties, final MessageQueue mq, final SendCallback sendCallback,
                          final long timeout) {
        Message message = createMessage(topic, tag, msgObj, properties);
        this.asyncSend(message, mq, sendCallback, timeout);
    }

    /**
     * 不关心是否送达到指定队列
     * @param message 消息体
     * @param messageQueue 消息队列
     */
    public void sendOneWay(final Message message, final MessageQueue messageQueue){
        try {
            producer.sendOneway(message, messageQueue);
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }
    /**
     * 不关心是否送达到指定队列
     *
     * @param topic      the topic
     * @param tag        the tag
     * @param msgObj     消息体
     * @param properties the properties
     * @param mq         the mq
     */
    public void sendOneWay(final String topic, final String tag, final Object msgObj,
            final Map<String, Object> properties, final MessageQueue mq) {
        Message message = createMessage(topic, tag, msgObj, properties);
        this.sendOneWay(message, mq);
    }

    /**
     * 同步发送到指定队列
     * @param message  消息体
     * @param selector selector
     * @param arg      对象
     * @return sendResult
     */
    public SendResult syncSend(final Message message, final MessageQueueSelector selector, final Object arg) {
        return syncSend(message, selector, arg, producer.getSendMsgTimeout());
    }

    /**
     * 同步发送到指定队列
     * @param message   消息体
     * @param selector  the selector
     * @param arg       对象
     * @param timeout   超时时间
     * @return sendResult
     */
    public SendResult syncSend(final Message message, final MessageQueueSelector selector, final Object arg,
                               final long timeout) {
        try {
            SendResult sendResult = producer.send(message, selector, arg, timeout);
            return sendResult;
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 同步发送消息到指定队列 （MessageQueueSelector）
     *
     * @param topic      the topic
     * @param tag        the tag
     * @param msgObj     消息体
     * @param properties the properties
     * @param selector   the selector
     * @param arg        the arg
     * @return sendResult
     */
    public SendResult syncSend(final String topic, final String tag, final Object msgObj,
                               final Map<String, Object> properties, final MessageQueueSelector selector, final Object arg) {
        return this.syncSend(topic, tag, msgObj, properties, selector, arg, producer.getSendMsgTimeout());
    }

    /**
     * 同步发送消息到指定队列 （MessageQueueSelector）
     *
     * @param topic     the topic
     * @param tag       the tag
     * @param msgObj    消息体
     * @param properties the properties
     * @param selector  the selector
     * @param arg       the args
     * @param timeout   超时时间
     * @return sendResult
     */
    public SendResult syncSend(final String topic, final String tag, final Object msgObj,
                               final Map<String, Object> properties, final MessageQueueSelector selector, final Object arg,
                               final long timeout) {
        Message message = createMessage(topic, tag, msgObj, properties);
        return this.syncSend(message, selector, arg, timeout);
    }

    /**
     * 异步发送到指定队列
     * @param message  消息体
     * @param selector selector
     * @param arg      对象
     * @param sendCallback 回调方法
     */
    public void asyncSend(final Message message, final MessageQueueSelector selector, final Object arg,
                          final SendCallback sendCallback) {
        this.asyncSend(message, selector, arg, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * 异步发送到指定队列
     * @param message   消息体
     * @param selector  selector
     * @param arg       对象
     * @param sendCallback 回调方法
     * @param timeout   超时时间
     */
    public void asyncSend(final Message message, final MessageQueueSelector selector, final Object arg,
                          final SendCallback sendCallback, final long timeout) {
        try {
            producer.send(message, selector, arg, sendCallback, timeout);
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 异步发送消息到指定队列 （MessageQueueSelector）
     *
     * @param topic      the topic
     * @param tag        the tag
     * @param msgObj     消息体
     * @param properties the properties
     * @param selector   the selector
     * @param arg        the arg
     * @param sendCallback 回调方法
     */
    public void asyncSend(final String topic, final String tag, final Object msgObj,
                          final Map<String, Object> properties, final MessageQueueSelector selector, final Object arg,
                          final SendCallback sendCallback) {
        this.asyncSend(topic, tag, msgObj, properties, selector, arg, sendCallback, producer.getSendMsgTimeout());
    }

    /**
     * 异步发送消息到指定队列 （MessageQueueSelector）
     *
     * @param topic       the topic
     * @param tag         the tag
     * @param msgObj      消息体
     * @param properties  the properties
     * @param selector    the selector
     * @param arg         the arg
     * @param sendCallback 回调方法
     * @param timeout     超时时间
     */
    public void asyncSend(final String topic, final String tag, final Object msgObj,
                          final Map<String, Object> properties, final MessageQueueSelector selector, final Object arg,
                          final SendCallback sendCallback, final long timeout) {
        Message message = createMessage(topic, tag, msgObj, properties);
        this.asyncSend(message, selector, arg, sendCallback, timeout);
    }

    /**
     * 不关心是否送达到指定队列
     * @param message  消息体
     * @param selector the selector
     * @param arg      the arg
     */
    public void sendOneWay(final Message message, final MessageQueueSelector selector, final Object arg) {
        try {
            producer.sendOneway(message, selector, arg);
        } catch (final Exception e) {
            log.error("send message failed. destination:{}:{}, message:{} ", message.getTopic(), message.getTags(), message);
            throw new MessagingException(e.getMessage(), e);
        }
    }

    /**
     * 不关心是否送达到指定队列
     *
     * @param topic   the topic
     * @param tag     the tag
     * @param msgObj  消息体
     * @param properties the properties
     * @param selector the selector
     * @param arg     the arg
     */
    public void sendOneWay(final String topic, final String tag, final Object msgObj,
                           final Map<String, Object> properties, final MessageQueueSelector selector, final Object arg) {
        Message message = createMessage(topic, tag, msgObj, properties);
        this.sendOneWay(message, selector, arg);
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(producer, "Property 'producer' is required");
        producer.start();
    }

    @Override
    public void destroy() throws Exception {
        Optional.ofNullable(producer).ifPresent(p -> p.shutdown());
    }

    private Message createMessage(final String topic, final String tag, final Object msgObj, final Map<String, Object> properties) {
        Message rocketMsg = new Message(topic, tag, serializer.serialize(msgObj));
        if (!CollectionUtils.isEmpty(properties)) {
            rocketMsg.setFlag((Integer) properties.getOrDefault("FLAG", 0));
            rocketMsg.setWaitStoreMsgOK((Boolean) properties.getOrDefault(MessageConst.PROPERTY_WAIT_STORE_MSG_OK, true));
            Optional.ofNullable((String) properties.get(MessageConst.PROPERTY_KEYS)).ifPresent(keys -> rocketMsg.setKeys(keys));
            Optional.ofNullable((Integer) properties.get(MessageConst.PROPERTY_DELAY_TIME_LEVEL)).ifPresent(delay -> rocketMsg.setDelayTimeLevel(delay));
            Optional.ofNullable((String) properties.get(MessageConst.PROPERTY_BUYER_ID)).ifPresent(buyerId -> rocketMsg.setBuyerId(buyerId));
            properties.entrySet().stream()
                    .filter(entry -> !MessageConst.STRING_HASH_SET.contains(entry.getKey()) && !Objects.equals(entry.getKey(), "FLAG"))
                    .forEach(entry -> rocketMsg.putUserProperty(entry.getKey(), String.valueOf(entry.getValue())));
        }
        return rocketMsg;
    }
}
